package yuekao8.machine;

import com.alibaba.alink.operator.batch.BatchOperator;
import com.alibaba.alink.operator.batch.dataproc.SplitBatchOp;
import com.alibaba.alink.operator.batch.evaluation.EvalBinaryClassBatchOp;
import com.alibaba.alink.operator.batch.source.CsvSourceBatchOp;
import com.alibaba.alink.operator.batch.source.MemSourceBatchOp;
import com.alibaba.alink.operator.batch.sql.FilterBatchOp;
import com.alibaba.alink.operator.common.evaluation.BinaryClassMetrics;
import com.alibaba.alink.pipeline.classification.*;
import com.alibaba.alink.pipeline.feature.Binning;
import org.apache.flink.types.Row;

public class GermanMachine {
    public static void main(String[] args) throws Exception {
        //（1）加载数据：使用Alink批处理加载数据集，封装数据集DataSet，提取特征features和标签label，前10条样本数据打印控制台；（4分）
        String filePath = "data/yk8/german.csv";
        String schema
                //f1,f4,f7,f10,f12,f15,f17,label
                //A11,6,A34,A43,1169,A65,A75,4,A93,A101,4,A121,67,A143,A152,2,A173,1,A192,A201,1
                = "f0 String,f1 int, f2 String, f3 String, f4 int, f5 String, f6 String, f7 int, f8 String, f9 String, f10 int" +
                ", f11 String, f12 int, f13 String, f14 String, f15 int, f16 String, f17 int, f18 String, f19 String, label int";
        CsvSourceBatchOp csvSource = new CsvSourceBatchOp()
                .setFilePath(filePath)
                .setSchemaStr(schema)
                .setFieldDelimiter(",")
                .setLenient(true)
                .setSkipBlankLine(true)
                .setIgnoreFirstLine(true);
//        csvSource.print();
        //封装数据集DataSet，提取特征 features 和标签 label
        String[] feature = new String[]{"f0", "f1", "f2", "f3", "f4", "f5", "f6", "f7", "f8", "f9", "f10", "f11", "f12", "f13", "f14", "f15", "f16", "f17", "f18", "f19"};
        String label = "label";
        //（2）数据清洗：进行合理的数据清洗。（4分）
        BatchOperator<?> op = new FilterBatchOp()
                .setClause("label=1 or label=2");
        BatchOperator<?> link = csvSource.link(op);
        //（3）特征选择：选择有价值的特征。（4分）
        //f1,f4,f7,f10,f12,f15,f17,label
        String[] features = new String[]{"f1", "f4", "f7", "f10", "f12", "f15", "f17"};
        //（4）数据集划分：按照8:2比例划分数据集：train训练集和test测试集，并查看条目数；（4分）
        BatchOperator<?> spliter = new SplitBatchOp().setFraction(0.8);
        BatchOperator<?> trainData = spliter.linkFrom(link);
        BatchOperator<?> testData = spliter.getSideOutput(0);
        System.out.println(trainData.count());
        System.out.println(testData.count());
        //（5）分箱：采用合适的分箱方法进行变量分箱（4分）
        //（6）Woe编码：将分箱值转换为woe值（4分）
        Binning binning = new Binning()
                .setEncode("WOE")
                .setSelectedCols("f0", "f2", "f3", "f5", "f6", "f8", "f9", "f11", "f13", "f14", "f16", "f18", "f19")
                .setLabelCol(label)
                .setPositiveLabelValueString("1");
        BatchOperator<?> transform = binning.fit(link).transform(link);
        transform.print();
        //（7）模型选择：选择三个分类模型，使用训练集训练模型。（4分）
        //决策树分类器 (DecisionTreeClassifier)
        DecisionTreeClassifier dec = new DecisionTreeClassifier()
                .setPredictionDetailCol("pred_detail")
                .setPredictionCol("pred")
                .setLabelCol("label")
                .setFeatureCols(features);

        //逻辑回归 (LogisticRegression)
        LogisticRegression lr = new LogisticRegression()
                .setPredictionDetailCol("pred_detail")
                .setFeatureCols(features)
                .setLabelCol("label")
                .setPredictionCol("pred");

        //朴素贝叶斯 (NaiveBayes)
        NaiveBayes ns = new NaiveBayes()
                .setPredictionDetailCol("pred_detail")
                .setFeatureCols(features)
                .setLabelCol("label")
                .setPredictionCol("pred");

        BatchOperator<?> dec_operator = dec.fit(trainData).transform(testData);
        BatchOperator<?> lr_operator = lr.fit(trainData).transform(testData);
        BatchOperator<?> ns_operator = ns.fit(trainData).transform(testData);
        //（8）模型评估：选择至少三个评估指标，评估模型。（4分）
        EvalBinaryClassBatchOp metrics = new EvalBinaryClassBatchOp()
                .setLabelCol("label")
                .setPredictionDetailCol("pred_detail");

        BinaryClassMetrics dec_binaryClassMetrics = metrics.linkFrom(dec_operator).collectMetrics();
        BinaryClassMetrics lr_binaryClassMetrics = metrics.linkFrom(lr_operator).collectMetrics();
        BinaryClassMetrics ns_binaryClassMetrics = metrics.linkFrom(ns_operator).collectMetrics();
        double dec_accuracy = dec_binaryClassMetrics.getAccuracy();
        double lr_accuracy = lr_binaryClassMetrics.getAccuracy();
        double ns_accuracy = ns_binaryClassMetrics.getAccuracy();
        System.out.println(dec_binaryClassMetrics.getAccuracy() + ":" + dec_binaryClassMetrics.getRecall() + ":" + dec_binaryClassMetrics.getPrecision());
        System.out.println(lr_binaryClassMetrics.getAccuracy() + ":" + lr_binaryClassMetrics.getRecall() + ":" + lr_binaryClassMetrics.getPrecision());
        System.out.println(ns_binaryClassMetrics.getAccuracy() + ":" + ns_binaryClassMetrics.getRecall() + ":" + ns_binaryClassMetrics.getPrecision());
        //（9）从三个训练好的模型中选择，效果最好的。（4分）
        if (dec_accuracy > lr_accuracy && dec_accuracy > ns_accuracy) {
            System.out.println("决策树分类器好");
        } else if (lr_accuracy > dec_accuracy && lr_accuracy > ns_accuracy) {
            System.out.println("逻辑回归好");
        } else {
            System.out.println("朴素贝叶斯好");
        }

        //（10）模型应用：模拟一条数据，使用训练好的模型预测。（4分）
    }
}
